package my.thesis.spoutsReliable;

import java.io.BufferedReader;
import java.io.IOException;
import java.io.InputStreamReader;
import java.util.HashMap;
import java.util.Map;

import java.net.URL;

import backtype.storm.spout.SpoutOutputCollector;
import backtype.storm.task.TopologyContext;
import backtype.storm.topology.OutputFieldsDeclarer;
import backtype.storm.topology.base.BaseRichSpout;
import backtype.storm.tuple.Fields;
import backtype.storm.tuple.Values;

public class HttpStreamReliable extends BaseRichSpout 
{
	private static final long serialVersionUID = 5773846193661629157L;
	private BufferedReader reader;
	private SpoutOutputCollector _collector;
		
	private String URL;
	Map<Integer, String> SpoutPendingTuples; //#non-acked tuples = #spout pending tuples
	Map<Integer, String> toSend;

	int i=1; //using for msgID

	public HttpStreamReliable (String connectionString)
	{
		this.URL=connectionString;
	}
	
	public void nextTuple() {
		try {
			String in;
			in = this.reader.readLine();
			if(in!=null)
			{
					SpoutPendingTuples.put(i, in);
					toSend.put(i, in);
					//Thread.sleep(100);
			}
			else 
			{
					//throw new ReportedFailedException("End Of file reached... bb");
					//Error(String message)
					//Constructs a new error with the specified detail message.
					//System.out.println("Nothing to read, wait 10 ms.");
					Thread.sleep(10);		
			}
			for(Map.Entry<Integer, String> entry : toSend.entrySet()){
				_collector.emit(new Values(entry.getValue()),entry.getKey()); //emit all the tuples from the toSend HashMap
			}
			toSend.clear(); //clear the HashMap
		} catch (Exception e){
			System.err.println("Error reading line "+e.getMessage());
			}
		i += 1;
	}

	public void open(Map conf, TopologyContext context,SpoutOutputCollector collector) {	
		this._collector = collector;
		
		this.SpoutPendingTuples = new HashMap<Integer, String>();
		this.toSend = new HashMap<Integer, String>();
		
		try 
		{
			URL url = new URL(URL);
			this.reader = new BufferedReader(new InputStreamReader(url.openStream()));	
		}
		catch (IOException e)
		{
			System.err.println("Error in communication with server ["+e.getMessage()+"]");
		}
	}
	
    public void ack(Object id) {
		System.out.println("OK ack for "+id+" Size : "+SpoutPendingTuples.size());
		SpoutPendingTuples.remove(id); //Remove the tuple acked from the Map	
    }

    public void fail(Object id) {
    	Integer msgID = (Integer) id;
    	toSend.put(msgID, SpoutPendingTuples.get(msgID)); //Resend the tuple that failed
    	System.err.println("fail for "+id);
    }

	public void declareOutputFields(OutputFieldsDeclarer declarer) {
		declarer.declare(new Fields("jsonString"));
	}

}
